package com.hugedata.idc.utils;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

public class HiveUtils
{
    private static final Logger LOGGER = Logger.getLogger(HiveUtils.class);

    /**
     * @param args
     *            hiveserver url,hql，user password，params...(name=value)
     * @throws ClassNotFoundException
     * @throws IOException
     * @throws SQLException 
     */
    public static void main(String[] args) throws ClassNotFoundException, IOException, SQLException
    {
        String hql = readContent(args[1]);
        hql = replaceParams(hql, args);
        LOGGER.info("hql script is:");
        LOGGER.info(hql);
        executeScript(args[0], hql, args[2], args[3]);
    }

    private static void executeScript(String url, String hql, String user, String password)
            throws ClassNotFoundException, SQLException
    {
        Class.forName("org.apache.hive.jdbc.HiveDriver");
        Connection con = null;
        Statement stmt = null;
        try
        {
            con = DriverManager.getConnection(url, user, password);
            stmt = con.createStatement();
            String[] split = hql.split(";");
            for (String subHQL : split)
            {
                if (StringUtils.isNotBlank(subHQL))
                {
                    LOGGER.info("execute hql " + subHQL);
                    stmt.execute(subHQL);
                }
            }
        }
        catch (SQLException e)
        {
            LOGGER.error(e);
            throw e; 
        }
        finally
        {
            if (stmt != null)
            {
                try
                {
                    stmt.close();
                }
                catch (SQLException e)
                {
                }
            }
            if (con != null)
            {
                try
                {
                    con.close();
                }
                catch (SQLException e)
                {
                }
            }
        }
    }

    private static String replaceParams(String hql, String[] args) throws IOException
    {
        for (int i = 4; i < args.length; i++)
        {
            String[] split = args[i].split("=");
            LOGGER.info("replace param:" + split[0]);
            Pattern p = Pattern.compile("(\\$\\{" + split[0] + "\\})");
            Matcher m = p.matcher(hql);
            hql = m.replaceAll(split[1]);
        }
        return hql;
    }

    private static String readContent(String hdfsFile) throws IOException
    {
        Configuration conf = HDFSUtils.getConf();
        FileSystem fs = null;
        FSDataInputStream fsDataInputStream = null;
        BufferedReader in = null;
        String hql = "";
        try
        {
            Path scriptFile = new Path(hdfsFile);
            fs = FileSystem.get(conf);
            fsDataInputStream = fs.open(scriptFile);
            InputStreamReader inputStreamReader = new InputStreamReader(fsDataInputStream, "UTF-8");
            in = new BufferedReader(inputStreamReader);
            String line;
            while ((line = in.readLine()) != null)
            {
                hql = hql.concat(line);
                hql = hql.concat(" ");
            }
        }
        finally
        {
            IOUtils.closeQuietly(in);
            IOUtils.closeQuietly(fsDataInputStream);
        }
        return hql;
    }
}
